package manager

import (
	"encoding/json"
	"fmt"
	"github.com/siddontang/go-mysql/mysql"
	"hermes/util"
	"io/ioutil"
	"net/http"
)

type TableFilterMessage struct {
	CareColumns []string
}

type allConfigInfos struct {
	data []configInfo	`json:"data"`
}

type configInfo struct {
	MysqlSlaveInfo mysqlSlaveInfo		`json:"connect"`
	LoadQueueInfo loadQueueInfo			`json:"queue"`
	TableInfoSchema []tableInfoSchema	`json:"tables"`
}

type mysqlSlaveInfo struct {
	SlaveId uint32   `json:"slave_id"`
	Host string      `json:"host"`
	Port uint16		 `json:"port"`
	Username string  `json:"username"`
	Password string  `json:"password"`
}

type loadQueueInfo struct {
	Type string				`json:"type"`
	Zookeeper []string		`json:"zookeeper"`
	BrokerList []string		`json:"broker_list"`
}

type tableInfoSchema struct {
	Table string 			 `json:"table"`
	CareColumns  []string    `json:"carecolumns"`
	SendColumns  []string    `json:"sendcolumns"`
}

type TableMetaData struct {
	Table string
	TableID uint64
	Position mysql.Position
	ColumnCount uint64
	ColumnNames  []string
	ColumnTypes  []byte
	ColumnTypeNames  []string
}

//private
var masterConfigInfo configInfo
var schemaFilterManager map[string](map[string]TableFilterMessage)
var dataxMasterName = "main"

//public
var SchemaMetaDataCache = make(map[string]map[uint64]TableMetaData)

func Init(chReload chan string)  {

	masterUrl:= "http://www.baidu.com"
	schemaFilterManager = map[string]map[string]TableFilterMessage{}
	initMaster(masterUrl)
	go waitForReload(chReload,masterUrl)

	//TODO need init SchemaMetaDataCache first by tsdb
	
}

func initMaster(masterUrl string)  {
	//reportSelf(masterUrl)
	//not get first?
	schemaFilterManager[dataxMasterName] = GetMasterConfigContent(masterUrl)
}

func reportSelf(masterUrl string)  {
	//TODO report self ip or hostname and port
}

func waitForReload(chReload chan string,masterUrl string)  {
	for {
		recv, ok := <-chReload
		if ok == false {
			fmt.Println("chan is closed")
			return
		}

		fmt.Println("recv change code:",recv)
		schemaFilterManager[dataxMasterName] = GetMasterConfigContent(masterUrl)
	}
}

func GetMasterConfigContent(url string) map[string]TableFilterMessage {
	resp,error := http.Get(url)
	if error !=nil{
		// handle error
	}
	defer resp.Body.Close()
	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		// handle error
	}
	fmt.Println(string(body))

	content := `
{
    "queue":{
        "type":"kafka",
        "zookeeper":[
            "10.3.5.30:2181"
        ],
        "broker_list":[
            "10.3.5.30:9092"
        ]
    },
    "tables":[
{"table":"b2b5.tb_merchandise","carecolumns":[],"sendcolumns":[]},
{"table":"b2b5.tb_speoffer_merchandise","carecolumns":[],"sendcolumns":[]},
{"table":"b2b5.tb_merchandise_stop_sale","carecolumns":[],"sendcolumns":[]},
{"table":"b2b5.tb_price","carecolumns":[],"sendcolumns":[]},
{"table":"b2b5.tb_merchandise_storage","carecolumns":[],"sendcolumns":[]},
{"table":"b2b5.tb_merchandise_ext","carecolumns":[],"sendcolumns":[]},
{"table":"b2b5.tb_merchandise_expiry_date","carecolumns":[],"sendcolumns":[]},
{"table":"b2b5.tb_general_agent","carecolumns":[],"sendcolumns":[]},
{"table":"b2b5.tb_brand_terminal","carecolumns":[],"sendcolumns":[]},
{"table":"b2b5.tb_merchandise_limitsale","carecolumns":[],"sendcolumns":[]},
{"table":"b2b5.tb_merchandise_label","carecolumns":[],"sendcolumns":[]},
{"table":"b2b5.tb_classify_merchandise","carecolumns":[],"sendcolumns":[]}
    ]
}`
	return  resolveContent(content)
}

func resolveContent(jsonContent string) map[string]TableFilterMessage{
	var configInfo configInfo
	var result map[string]TableFilterMessage
	if err := json.Unmarshal([]byte(jsonContent), &configInfo); err == nil {
		masterConfigInfo = configInfo
		fmt.Println(configInfo)
		result = merge(configInfo.TableInfoSchema)
	} else {
		fmt.Println(err)
	}

	return result
}

func merge(tabileInfoList []tableInfoSchema) map[string]TableFilterMessage{
	result:= map[string]TableFilterMessage{}
	careColumnsMapH:=map[string] string{}
	careColumnsMap:=map[string][]string{}
	for _,tableItem:= range tabileInfoList {
		if len(tableItem.CareColumns) == 0 {
			result[tableItem.Table] = TableFilterMessage{}
		}
		for _,careColum:= range tableItem.CareColumns {
			careColumnsMapH[careColum] = tableItem.Table
		}
	}

	for k,v := range careColumnsMapH {
		if _, ok := careColumnsMap[v]; !ok {
			careColumnsMap[v] =[]string{}
		}
		careColumnsMap[v] = append(careColumnsMap[v], k)
	}

	for k,v := range careColumnsMap {
		result[k] = TableFilterMessage{CareColumns:v}
	}

	return result
}

func GetMysqlConfig() mysqlSlaveInfo {
	//masterConfigInfo.MysqlSlaveInfo
	mysqlConfig:=util.GetAppConfig().MysqlConfig
	return mysqlSlaveInfo{mysqlConfig.SlaveId,
		mysqlConfig.Host,mysqlConfig.Port,
		mysqlConfig.Username,
		mysqlConfig.Password}
}

func GetLoadConfig() loadQueueInfo {
	return masterConfigInfo.LoadQueueInfo
}

func IsTableFilter(tableName string) bool {
	_,ok := schemaFilterManager[dataxMasterName][tableName]
	if ok {
		return false
	}
	return true
}

func GetTableFilterMessage(tableName string) TableFilterMessage {
	tableFilterMessages,ok := schemaFilterManager[dataxMasterName][tableName]
	if ok {
		return tableFilterMessages
	}
	return TableFilterMessage{}
}

func GetTableMetaData(tableName string,tableId uint64) TableMetaData {
	//tableMetaDatas := SchemaMetaDataCache[tableName]
	//var lastTableMetaData TableMetaData
	//for _,tableMetaData :=range tableMetaDatas{
	//	if position.Compare(tableMetaData.Position) < 0{
	//		return lastTableMetaData
	//	}
	//	lastTableMetaData = tableMetaData
	//}
	//return lastTableMetaData
	return SchemaMetaDataCache[tableName][tableId]
}